package DataStreamApi.Source读取;


import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 自动生成数据的测试工具
 */
public class Flink04_DATAGEN {
    public static void main(String[] args) throws Exception {

        /**
         * 创建执行环境
         * IDEA运行的时候，
         * 需要导入依赖
         */
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        /**
         * 数据生成器Source,最大构造方法四个参数
         * 第一个，GeneratorFunction接口，需要实现，重写map方法
         * 第二个，Longf类型，自动生成的最大条数
         * 第三个，限速策略，比如每秒生成了多少数据
         *第四个，返回的数据类型
         */
        DataGeneratorSource<String> stringDataGeneratorSource = new DataGeneratorSource<>(new GeneratorFunction<Long, String>() {
            @Override
            public String map(Long value) throws Exception {
                return "number" + value;
            }
        }, 1000000000, RateLimiterStrategy.perSecond(1000), Types.STRING);


        DataStreamSource<String> flink13_datagen = executionEnvironment.fromSource(stringDataGeneratorSource, WatermarkStrategy.noWatermarks(), "Flink13_DATAGEN");

        flink13_datagen.print();


        executionEnvironment.execute();


    }
}
